/**
 * Project Reactor
 */


#include "EventLoop.h"

/**
 * EventLoop implementation
 */
// void display(map<int,TcpConnection> &map){
//     for(auto it = map.begin() ; it != map.end() ; it++){
//         cout<<it->first<<":"<<it->second.getSock().getFd()<<endl;
//     }
// }

/**
 * @param accept
 */
EventLoop::EventLoop(Acceptor & acceptor) 
:m_epfd(createEpollFd())
,m_acceptor(acceptor)
,m_evtList(MAX_EVTLIST_SIZE)
,m_isLooping(false)
,m_connMap()
,m_evtfd(createEventFd())
,m_pendings()
,m_mutex()
{

}

EventLoop::~EventLoop() {
    close(m_epfd);
}

void EventLoop::loop() {
    int listenfd = m_acceptor.getSock().getFd();
    addEpollReadFd(listenfd);
    addEpollReadFd(m_evtfd);
    m_isLooping = true;
    while(m_isLooping){
        waitEpollFd();
    }
}

void EventLoop::handleNewConnection() {
    int connfd = m_acceptor.m_accept();
    addEpollReadFd(connfd);

    TcpConnectionPtr con(new TcpConnection(connfd,this));
    m_connMap[connfd] = con;

    con->setNewConnectionCallback(std::move(m_onNewConnection));
    con->setMessageCallback(std::move(m_onMessage));
    con->setCloseCallback(std::move(m_onClose));

    cout<<"建立新连接，连接fd = "<<connfd<<endl;
    con->handleNewConnectionCallback();
}

/**
 * @param fd
 */

void EventLoop::handleMessage(int fd) {
    auto it = m_connMap.find(fd); // 获取迭代器
    if (it == m_connMap.end()) {
        cout<<"连接不存在"<<endl;
        return; // 键不存在时直接返回
    }
    bool flag = it->second->isClosed();
    cout<<"连接状态："<<flag<<endl;
    if(flag){
        it->second->handleCloseCallback();
        delEpollReadFd(fd);
        m_connMap.erase(it);
    }else{
        // cout<<"本次响应信息："<<it->second->streamToString()<<endl;
        // string msg = it->second->receive();
        // cout<<"收到消息："<<msg<<endl;
        // it->second->send("hello baby\n");
        it->second->handleMessageCallback();
    }

}


/**
 * @return int
 */
int EventLoop::createEpollFd() {
    m_epfd = epoll_create(MAX_EVTLIST_SIZE);
    if(-1 == m_epfd){
        perror("epoll_create");
        return -1;
    }
    return m_epfd;
}

/**
 * @param fd
 */
void EventLoop::addEpollReadFd(int fd) {
    struct epoll_event evt,ep[MAX_EVTLIST_SIZE];
    evt.events = EPOLLIN;
    evt.data.fd = fd;
    int ret = epoll_ctl(m_epfd,EPOLL_CTL_ADD,fd,&evt);
    if(ret<0){
        perror("epoll_ctl");
        return;
    }
    cout<<"添加成功"<<endl;
}

/**
 * @param fd
 */
void EventLoop::delEpollReadFd(int fd) {
    struct epoll_event evt;
    evt.data.fd = fd;
    int ret = epoll_ctl(m_epfd,EPOLL_CTL_DEL,fd,&evt);
    if(ret <0){
        perror("epoll_del");
        return;
    }
}

// int EventLoop::m_epoll_wait() {
//     int readyNum = epoll_wait(m_epfd,m_evtList,MAX_EVTLIST_SIZE,-1);
//     return readyNum;
// }


void EventLoop::waitEpollFd() {
    int readyNum = 0;
    do{
        readyNum = epoll_wait(m_epfd,m_evtList.data(),m_evtList.size(),-1);
    }while(-1 == readyNum &&errno == EINTR);
    if(-1 == readyNum){
        std::cerr<<"-1 == readyNum"<<endl;
    }else if(0 == readyNum){
        cout<<"epoll_wait timeout"<<endl;
    }else{
        if(readyNum == (int)m_evtList.size()){
            m_evtList.resize(2*readyNum);
        }
        for(int i = 0 ; i< readyNum ; i++){
            int listenfd = m_acceptor.getSock().getFd();
            int fd = m_evtList[i].data.fd;
            if(fd == listenfd){
                handleNewConnection();
                // display(m_connMap);
            }else if(fd == m_evtfd){
                handleRead();
                doPendingFunctors();
            }else{
                handleMessage(fd);
            }
        }
    }
}


void EventLoop::unloop() {
    m_isLooping = false;
}

void EventLoop::setNewConnectionCallback(TcpConnectionCallback&& tcc){
    m_onNewConnection = std::move(tcc);
}
    
/**
 * @param tcc
 */
void EventLoop::setMessageCallback(TcpConnectionCallback&& tcc){
    m_onMessage = std::move(tcc);
}
    
/**
 * @param tcc
 */
void EventLoop::setCloseCallback(TcpConnectionCallback&& tcc){
    m_onClose = std::move(tcc);
}

int EventLoop::createEventFd(){
    int fd = eventfd(0,0);
    if(fd < 0){
        perror("createEventFd");
        return -1;
    }
    return fd;
}
    
void EventLoop::handleRead(){
    uint64_t val = 1;
    ssize_t ret = read(m_evtfd,&val,sizeof(uint64_t));
    if(ret != sizeof(uint64_t)){
        perror("handleRead");
        return;
    }
}
    
void EventLoop::wakeup(){
    uint64_t val = 1;
    ssize_t ret = write(m_evtfd,&val,sizeof(uint64_t));
    if(ret != sizeof(uint64_t)){
        perror("wakeup");
        return;
    }
}
    
void EventLoop::doPendingFunctors(){
    vector<Functor> temp;
    {
        lock_guard<mutex> lg(m_mutex);
        temp.swap(m_pendings);
    }
    for(auto &cb : temp){
        cb();
    }
}
    
/**
 * @param cb
 */
void EventLoop::runInLoop(Functor && cb){
    {
        lock_guard<mutex> lg(m_mutex);
        m_pendings.push_back(std::move(cb));
    }
    
    wakeup();
}